#include "Stdafx.h"
#include "vcclr.h"

#include "ZmqSocket.h"
#include "ZmqContext.h"

using namespace System;
using namespace System::Threading;
using namespace vJine::Core::IO::Bin;

namespace vJine {
	namespace ZMQ {
		ZmqRole ZmqSocket::Role::get() {
			return this->_Role;
		}
		void ZmqSocket::Role::set(ZmqRole value) {
			this->_Role = value;
		}

		ZmqSocket::ZmqSocket(void) {
		}

		ZmqSocket::ZmqSocket(void* context, ZmqRole role)
		{
			this->socket = 
				zmq_socket(context, (int)role);
			if(this->socket == NULL) {
				ZmqContext::throw_ex("ZmqSocket.ctor(void* context, ZmqRole role)");
			}

			this->context = context;
			this->Role = role;

			this->message = new zmq_msg_t();
			zmq_msg_init(this->message);

			if(role == ZmqRole::SUB) {
				zmq_setsockopt(this->socket, ZMQ_SUBSCRIBE, "", 0);
			}

			this->buffer = gcnew array<byte>(1024);
			this->cache = this->initCahce(this->cache);
		}

		ZmqSocket::~ZmqSocket() {

			zmq_close(this->socket);
				delete this->socket;

			zmq_msg_close(this->message);
			delete this->message;
		}

		ZmqSocket^ ZmqSocket::Bind(String^ endpoint) {
			
			Monitor::Enter(this);
			try {
				if(zmq_bind(this->socket, this->convert_string_to_chars(endpoint)) == -1) {
					ZmqContext::throw_ex("ZmqSocket.Bind(string endpoint)");
				}

				return this;
			} finally {
				Monitor::Exit(this);
			}
		}

		ZmqSocket^ ZmqSocket::Connect(String^ endpoint) {

			Monitor::Enter(this);
			try {
				if(zmq_connect(this->socket, this->convert_string_to_chars(endpoint)) == -1) {
					ZmqContext::throw_ex("ZmqSocket.Connect(string endpoint)");
				}

				return this;
			} finally {
				Monitor::Exit(this);
			}
		}

		ZmqSocket^ ZmqSocket::Subscribe(String^ topic) {
			
			Monitor::Enter(this);
			try {
				this->set(ZMQ_SUBSCRIBE, topic);

				return this;
			} finally {
				Monitor::Exit(this);
			}
		}

		ZmqSocket^ ZmqSocket::UnSubscribe(String^ topic) {

			Monitor::Enter(this);
			try {
				this->set(ZMQ_UNSUBSCRIBE, topic);

				return this;
			} finally {
				Monitor::Exit(this);
			}
		}

		void ZmqSocket::Close() {

			Monitor::Enter(this);
			try {
				if (zmq_close(this->socket) != 0) {
					ZmqContext::throw_ex("ZmqSocket.Close()");
				}
			} finally {
				Monitor::Exit(this);
			}
		}

		generic <typename T>
		void ZmqSocket::Send(T data) {
			if(data == nullptr) {
				ZmqContext::throw_ex("ZmqSocket.Send<T>(T data)","arg.data shouldn't be null.");
			}

			Monitor::Enter(this);
			try {
				this->cache = 
					this->initCahce(this->cache);
				BinHelper<T>::ToString(data, this->cache);

				this->cache->Position = 0;
				this->Send((Stream^)this->cache, false);
			} finally {
				Monitor::Exit(this);
			}
		}

		void ZmqSocket::Send(array<byte>^ data) {
			if(data == nullptr) {
				ZmqContext::throw_ex("ZmqSocket.Send(byte[] data)","arg.data shouldn't be null.");
			}

			this->Send(data, 0, data->Length, false);
		}

		void ZmqSocket::Send(array<byte>^ data, bool more) {
			if(data == nullptr) {
				ZmqContext::throw_ex("ZmqSocket.Send(byte[] data, bool more)","arg.data shouldn't be null.");
			}

			this->Send(data, 0, data->Length, more);
		}

		void ZmqSocket::Send(array<byte>^ data, int len) {
			if(data == nullptr) {
				ZmqContext::throw_ex("ZmqSocket.Send(byte[] data, int len)","arg.data shouldn't be null.");
			}

			this->Send(data, 0, len, false);
		}

		void ZmqSocket::Send(array<byte>^ data, int len, bool more) {
			if(data == nullptr) {
				ZmqContext::throw_ex("ZmqSocket.Send(byte[] data, int len, bool more)","arg.data shouldn't be null.");
			}

			this->Send(data, 0, len, more);
		}

		void ZmqSocket::Send(array<byte>^ data, int offset, int len) {
			if(data == nullptr) {
				ZmqContext::throw_ex("ZmqSocket.Send(byte[] data, int offset, int len)","arg.data shouldn't be null.");
			}

			this->Send(data, offset, len, false);
		}

		void ZmqSocket::Send(array<byte>^ data, int offset, int len, bool more) {
			
			Monitor::Enter(this);
			try{
				String^ fcn_name = "ZmqSocket.Send(byte[] data, int offset, int len, bool more)";
				if(data == nullptr) {
					ZmqContext::throw_ex(fcn_name,"arg.data shouldn't be null.");
				} else if (offset < 0) {
					ZmqContext::throw_ex(fcn_name,"arg.offset shouldn't less than 0.");
				} else if (len <= 0) {
					ZmqContext::throw_ex(fcn_name,"arg.len should bigger than 0.");
				} else if (len + offset > data->Length) {
					ZmqContext::throw_ex(fcn_name,"arg.offset + arg.len shouldn't bigger than data.length.");
				} else {
					pin_ptr<byte> ptr = &data[0];
					if (zmq_send(this->socket, ptr, len, (more ? ZMQ_SNDMORE : 0)) == -1) {
						ZmqContext::throw_ex(fcn_name);
					}
				}
			} finally {
				Monitor::Exit(this);
			}
		}

		void ZmqSocket::Send(Stream^ stream) {
			this->Send(stream, false);
		}

		void ZmqSocket::Send(Stream^ stream, bool more) {
			String^ fcn_name = "ZmqSocket.Send(Stream stream, bool more)";
			if(stream == nullptr) {
				ZmqContext::throw_ex(fcn_name,"arg.stream shouldn't be null.");
			}

			Monitor::Enter(this);
			try {
				int buffer_size = this->buffer->Length;
				pin_ptr<byte> ptr = &this->buffer[0];
				for(;;) {
					int lenRead = stream->Read(buffer, 0, buffer_size);

					if(lenRead == buffer_size) {
						if (zmq_send(this->socket, ptr, buffer_size, ZMQ_SNDMORE)) {
							ZmqContext::throw_ex(fcn_name);
						}
					} else {
						if (zmq_send(this->socket, ptr, lenRead, (more ? ZMQ_SNDMORE : 0)) == -1) {
							ZmqContext::throw_ex(fcn_name);
						}
						break;
					}
				}
			} finally {
				Monitor::Exit(this);
			}
		}

		generic <typename T>
		T ZmqSocket::Receive() {
			
			Monitor::Enter(this);
			try {
				this->cache = 
					this->initCahce(this->cache);
				this->Receive((Stream^)this->cache);

				this->cache->Position = 0;
				return BinHelper<T>::Parse(this->cache);
				
			} finally {
				Monitor::Exit(this);
			}

		}

		generic <typename T>
		void ZmqSocket::Receive(T data) {

			Monitor::Enter(this);
			try {
				this->cache = 
					this->initCahce(this->cache);
				this->Receive((Stream^)this->cache);

				this->cache->Position = 0;
				BinHelper<T>::Parse(this->cache, data);
				
			} finally {
				Monitor::Exit(this);
			}

		}


		int ZmqSocket::Receive(array<byte>^ data) {

			if(data == nullptr) {
				ZmqContext::throw_ex("ZmqSocket.Receive(byte[] data)","arg.data shouldn't be null.");
			}

			bool more = false;
			int offset = 0; 
			int len = data->Length;
			int lenReceive = 0;

			Monitor::Enter(this);
			try {
				do{
					lenReceive = this->Receive(data, offset, len, more);
					offset += lenReceive;
					len -= lenReceive;
				} while (more);

				return data->Length - len;
			} finally {
				Monitor::Exit(this);
			}
		}

		int ZmqSocket::Receive(array<byte>^ data, [Out] bool% more) {

			String^ fcn_name = "ZmqSocket.Receive(byte[] data, out bool more)";
			if(data == nullptr) {
				ZmqContext::throw_ex(fcn_name,"arg.data shouldn't be null.");
			}

			return this->Receive(data, 0, data->Length, more);
		}

		int ZmqSocket::Receive(array<byte>^ data, int len, [Out] bool% more) {

			String^ fcn_name = "ZmqSocket.Receive(byte[] data, int offset, int len, out bool more)";
			if(data == nullptr) {
				ZmqContext::throw_ex(fcn_name,"arg.data shouldn't be null.");
			} else if (len <= 0) {
				ZmqContext::throw_ex(fcn_name,"arg.len should bigger than 0.");
			}

			return this->Receive(data, 0, len, more);
		}

		int ZmqSocket::Receive(array<byte>^ data, int offset, int len, [Out] bool% more) {

			String^ fcn_name = "ZmqSocket.Receive(byte[] data, int offset, int len, out bool more)";
			if(data == nullptr) {
				ZmqContext::throw_ex(fcn_name,"arg.data shouldn't be null.");
			} else if (offset < 0) {
				ZmqContext::throw_ex(fcn_name,"arg.offset shouldn't less than 0.");
			} else if (len <= 0) {
				ZmqContext::throw_ex(fcn_name,"arg.len should bigger than 0.");
			} else if (len + offset > data->Length) {
				ZmqContext::throw_ex(fcn_name,"arg.offset + arg.len shouldn't bigger than data.length.");
			}

			Monitor::Enter(this);
			try {
				int lenRemain = len;
				bool first_loop = true;
				for(;;) {
					int lenCache = 
						this->cache->Length - this->cache->Position;
					
					if(lenCache >= lenRemain) {
						this->cache->Read(data, offset, lenRemain);
						return len;
					} else if (lenCache > 0) {
						this->cache->Read(data, offset, lenCache);
						offset += lenCache;
						lenRemain -= lenCache;

						this->cache =
							this->initCahce(this->cache);
					}
					
					if(!first_loop && !more) {
						return len - lenRemain;
					} else {
						int lastPosition = this->cache->Position;
						this->Receive((Stream^)this->cache, more);
						this->cache->Position = lastPosition;
					}
					first_loop = false;
				}			
				
			} finally {
				Monitor::Exit(this);
			}

		}

		int ZmqSocket::Receive(Stream^ stream) {

			String^ fcn_name = "ZmqSocket.Receive(Stream^ stream, [Out] bool% more) ";
			if(stream == nullptr) {
				ZmqContext::throw_ex(fcn_name,"arg.stream shouldn't be null.");
			}
			
			Monitor::Enter(this);
			try {
				bool more = false; int lenRead = 0;
				do {
					lenRead += this->Receive(stream, more);
				} while (more);

				return lenRead;
			} finally {
				Monitor::Exit(this);
			}
		}

		int ZmqSocket::Receive(Stream^ stream, [Out] bool% more) {

			String^ fcn_name = "ZmqSocket.Receive(Stream stream, [Out] bool% more) ";
			if(stream == nullptr) {
				ZmqContext::throw_ex(fcn_name,"arg.stream shouldn't be null.");
			}

			Monitor::Enter(this);
			try {
				if(zmq_msg_recv(this->message, this->socket, 0) == -1) {
					ZmqContext::throw_ex(fcn_name);
				}

				more = zmq_msg_more(this->message) == 1;
				int len_msg = zmq_msg_size(this->message);

				int buffer_size = this->buffer->Length;
				unsigned char* ptr_msg = (unsigned char*)zmq_msg_data(this->message);
				for(int i = 0; i < len_msg; i+= buffer_size) {
					int len = 
						i + buffer_size >= len_msg ? len_msg - i : buffer_size;
					if(len == 0) {
						break;
					}

					Marshal::Copy(IntPtr(ptr_msg), this->buffer, i, len);
					stream->Write(this->buffer, 0, len);
				}
				
				return len_msg;
			} finally {
				Monitor::Exit(this);
			}
		}

		bool ZmqSocket::available::get() {

			Monitor::Enter(this);
			try {
				zmq_pollitem_t items[1];
				items[0].socket = socket;
				items[0].events = ZMQ_POLLIN;
				
				int result = zmq_poll(items, 1, 1);
				if(result == -1) {
					ZmqContext::throw_ex("ZmqSocket.available.get()");
				}

				return result > 0;
			} finally {
				Monitor::Exit(this);
			}
		}

		int ZmqSocket::reconnect_ivl::get() {
			return this->get(ZMQ_RECONNECT_IVL);
		}
		void ZmqSocket::reconnect_ivl::set(int value) {
			this->set(ZMQ_RECONNECT_IVL_MAX, value);
		}

		int ZmqSocket::reconnect_ivl_max::get() {
			return this->get(ZMQ_RECONNECT_IVL_MAX);
		}
		void ZmqSocket::reconnect_ivl_max::set(int value) {
			this->set(ZMQ_RECONNECT_IVL, value);
		}

		int ZmqSocket::send_hwm::get() {
			return this->get(ZMQ_SNDHWM);
		}
		void ZmqSocket::send_hwm::set(int value) {
			this->set(ZMQ_SNDHWM, value);
		}

		int ZmqSocket::send_tmo::get() {
			return this->get(ZMQ_SNDTIMEO);
		}
		void ZmqSocket::send_tmo::set(int value) {
			this->set(ZMQ_SNDTIMEO, value);
		}

		int ZmqSocket::send_buff::get() {
			return this->get(ZMQ_SNDBUF);
		}
		void ZmqSocket::send_buff::set(int value) {
			this->set(ZMQ_SNDBUF, value);
		}

		int ZmqSocket::recv_hwm::get() {
			return this->get(ZMQ_RCVHWM);
		}
		void ZmqSocket::recv_hwm::set(int value) {
			this->set(ZMQ_RCVHWM, value);
		}

		int ZmqSocket::recv_tmo::get() {
			return this->get(ZMQ_RCVTIMEO);
		}
		void ZmqSocket::recv_tmo::set(int value) {
			this->set(ZMQ_RCVTIMEO, value);
		}

		int ZmqSocket::recv_buff::get() {
			return this->get(ZMQ_RCVBUF);
		}
		void ZmqSocket::recv_buff::set(int value) {
			this->set(ZMQ_RCVBUF, value);
		}

		bool ZmqSocket::recv_more::get() {
			return this->get(ZMQ_RCVBUF) == 1;
		}

		int64_t ZmqSocket::max_size::get() {
			return this->getl(ZMQ_MAXMSGSIZE);
		}
		void ZmqSocket::max_size::set(int64_t value) {
			this->setl(ZMQ_MAXMSGSIZE, value);
		}

		int ZmqSocket::linger::get() {
			return this->get(ZMQ_LINGER);
		}
		void ZmqSocket::linger::set(int value) {
			this->set(ZMQ_LINGER, value);
		}

		int ZmqSocket::backlog::get() {
			return this->get(ZMQ_BACKLOG);
		}
		void ZmqSocket::backlog::set(int value) {
			this->set(ZMQ_BACKLOG, value);
		}

		int ZmqSocket::hops::get() {
			return this->get(ZMQ_MULTICAST_HOPS);
		}
		void ZmqSocket::hops::set(int value) {
			this->set(ZMQ_MULTICAST_HOPS, value);
		}

		int ZmqSocket::rate::get() {
			return this->get(ZMQ_RATE);
		}
		void ZmqSocket::rate::set(int value) {
			this->set(ZMQ_RATE, value);
		}

		int ZmqSocket::recovery_ivl::get() {
			return this->get(ZMQ_RECOVERY_IVL);
		}
		void ZmqSocket::recovery_ivl::set(int value) {
			this->set(ZMQ_RECOVERY_IVL, value);
		}

		bool ZmqSocket::ipv6::get() {
			return this->get(ZMQ_MULTICAST_HOPS) == 1;
		}
		void ZmqSocket::ipv6::set(bool value) {
			this->set(ZMQ_MULTICAST_HOPS, (value ? 1 : 0));
		}

		int ZmqSocket::keep_alive::get() {
			return this->get(ZMQ_TCP_KEEPALIVE);
		}
		void ZmqSocket::keep_alive::set(int value) {
			this->set(ZMQ_TCP_KEEPALIVE, value);
		}

		int ZmqSocket::keep_alive_cnt::get() {
			return this->get(ZMQ_TCP_KEEPALIVE_CNT);
		}
		void ZmqSocket::keep_alive_cnt::set(int value) {
			this->set(ZMQ_TCP_KEEPALIVE_CNT, value);
		}

		int ZmqSocket::keep_alive_ivl::get() {
			return this->get(ZMQ_TCP_KEEPALIVE_INTVL);
		}
		void ZmqSocket::keep_alive_ivl::set(int value) {
			this->set(ZMQ_TCP_KEEPALIVE_INTVL, value);
		}

		bool ZmqSocket::plain_server::get() {
			return this->get(ZMQ_PLAIN_SERVER) == 1;
		}
		void ZmqSocket::plain_server::set(bool value) {
			this->set(ZMQ_PLAIN_SERVER, (value ? 1 : 0));
		}

		String^ ZmqSocket::plain_user::get() {
			return this->get(ZMQ_PLAIN_USERNAME, 100);
		}
		void ZmqSocket::plain_user::set(String^ value) {
			this->set(ZMQ_PLAIN_USERNAME, value);
		}

		String^ ZmqSocket::plain_password::get() {
			return this->get(ZMQ_PLAIN_PASSWORD, 100);
		}
		void ZmqSocket::plain_password::set(String^ value) {
			this->set(ZMQ_PLAIN_PASSWORD, value);
		}

		String^ ZmqSocket::curve_serverkey::get() {
			return this->get(ZMQ_CURVE_SERVERKEY, 41);
		}
		void ZmqSocket::curve_serverkey::set(String^ value) {
			this->set(ZMQ_CURVE_SERVERKEY, value);
		}

		String^ ZmqSocket::curve_publickey::get() {
			return this->get(ZMQ_CURVE_PUBLICKEY, 41);
		}
		void ZmqSocket::curve_publickey::set(String^ value) {
			this->set(ZMQ_CURVE_PUBLICKEY, value);
		}

		String^ ZmqSocket::curve_secretkey::get() {
			return this->get(ZMQ_CURVE_SECRETKEY, 41);
		}
		void ZmqSocket::curve_secretkey::set(String^ value) {
			this->set(ZMQ_CURVE_SECRETKEY, value);
		}

		bool ZmqSocket::conflate::get() {
			return this->get(ZMQ_CONFLATE) == 1;
		}
		void ZmqSocket::conflate::set(bool value) {
			this->set(ZMQ_CONFLATE, (value ? 1 : 0));
		}

		String^ ZmqSocket::last_endpoint::get() {
			return this->get(ZMQ_LAST_ENDPOINT, 100)->Trim();
		}

		void ZmqSocket::set(int opt_name, int value) {

			Monitor::Enter(this);
			try{
				if(zmq_setsockopt(this->socket, (int)(opt_name), &value, sizeof(int)) == -1) {
					ZmqContext::throw_ex("ZmqSocket.set(int opt_name, int value)");
				}
			} finally {
				Monitor::Exit(this);
			}
		}

		void ZmqSocket::setl(int opt_name, int64_t value) {

			Monitor::Enter(this);
			try{
				if(zmq_setsockopt(this->socket, (int)(opt_name), &value, sizeof(int64_t)) == -1) {
					ZmqContext::throw_ex("ZmqSocket.set(int opt_name, int value)");
				}
			} finally {
				Monitor::Exit(this);
			}
		}

		void ZmqSocket::set(int opt_name, String^ value) {

			Monitor::Enter(this);
			try{
				char* pvalue = this->convert_string_to_chars(value);
				if(zmq_setsockopt(this->socket, (int)(opt_name), pvalue, value->Length) == -1) {
					ZmqContext::throw_ex("ZmqSocket.set(int opt_name, String^ value)");
				}
			} finally {
				Monitor::Exit(this);
			}

		}

		int ZmqSocket::get(int opt_name) {

			Monitor::Enter(this);
			try{
				int value = 0; size_t len = sizeof(int);

				if(zmq_getsockopt(this->socket, (int)opt_name, &value, &len) == -1) {
					ZmqContext::throw_ex("ZmqSocket.get(int opt_name)");
				}
				return value;
			} finally {
				Monitor::Exit(this);
			}

		}

		int64_t ZmqSocket::getl(int opt_name) {

			Monitor::Enter(this);
			try{
				int value = 0; size_t len = sizeof(int64_t);

				if(zmq_getsockopt(this->socket, (int)opt_name, &value, &len) == -1) {
					ZmqContext::throw_ex("ZmqSocket.get(int opt_name)");
				}
				return value;
			} finally {
				Monitor::Exit(this);
			}

		}

		String^ ZmqSocket::get(int opt_name, int len) {

			Monitor::Enter(this);
			try{
				size_t size = (sizeof(char) * len);
				char* value = (char*)malloc(size);
				if(zmq_getsockopt(this->socket, (int)opt_name, value, &size) == -1) {
					ZmqContext::throw_ex("ZmqSocket.get(int opt_name, int len)");
				}

				return gcnew String(value);
			} finally {
				Monitor::Exit(this);
			}
		}

		MemoryStream^ ZmqSocket::initCahce(MemoryStream^ cache) {
			Monitor::Enter(this);
			try{
				if(cache == nullptr) {
					return gcnew MemoryStream();
				} else {
					cache->Position = 0;
					cache->SetLength(0);
					return cache;
				}
			} finally {
				Monitor::Exit(this);
			}
		}

		char* ZmqSocket::convert_string_to_chars(String^ addr) {
			if(String::IsNullOrEmpty(addr)) {
				return NULL;
			}

			pin_ptr<const wchar_t> addr_p = PtrToStringChars(addr);
			array<char>^ addr_2 = gcnew array<char>(addr->Length);
			pin_ptr<char> addr_p_2 = &addr_2[0];
			WideCharToMultiByte(CP_ACP, 0, addr_p,-1, addr_p_2,addr->Length, 0, 0);
			return addr_p_2;
		}
	}
}
